package com.kafka.producer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import net.sf.json.JSONObject;

import com.kafka.bean.User;
import com.kafka.common.Constant;
import com.kafka.common.String.StringValues;
import com.kafka.core.BaseProducer;
import com.kafka.dao.IccardMapper;
import com.kafka.dao.UserMapper;
import com.kafka.enumtype.MsgType;
import com.kafka.enumtype.ResponseType;
import com.kafka.producer.inter.IProducer;
import com.kafka.util.BaseUtil;
import com.kafka.valid.UserValid;

/**
 * 注册消息类
 * 
 * @author leihongan
 * 
 */
public class UserProducer extends BaseProducer implements IProducer {
	UserMapper userDao = sqlSession.getMapper(UserMapper.class);
	IccardMapper iccardDao = sqlSession.getMapper(IccardMapper.class);
	private final Producer<String, String> producer;
	public String TOPIC;
	public UserProducer() {
		Properties properties = new Properties();
		try {
			properties.load(ClassLoader
					.getSystemResourceAsStream("user.properties"));
			TOPIC = properties.getProperty("topic");
		} catch (IOException e) {
			e.printStackTrace();
		}
		producer = new Producer<String, String>(new ProducerConfig(properties));
	}

	public String produce(JSONObject message) throws Exception {
		if(MsgType.REGISTER.toString().equals(message.getString("msgtype"))){
			return registerProduce(message);
		} else if(MsgType.UDATE.toString().equals(message.getString("msgtype"))){
			return updateProduce(message);
		} else if(MsgType.ACTIVATEUSER.toString().equals(message.getString("msgtype"))){
			return activateUserProduce(message);
		} else if(MsgType.USERBLACKLIST.toString().equals(message.getString("msgtype"))){
			return blackListProduce(message);
		} else if(MsgType.CLOSEUSER.toString().equals(message.getString("msgtype"))){
			return closeUserProduce(message);
		} else{
			throw new Exception("no match funtion!");
		}
		
	}

	/**
	 * 注册处理
	 * @param message
	 * @return
	 */
	private String registerProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = UserValid.validRegister(rebuildJSON(message.getJSONObject("datas")), msgId,Constant.SERVICE);
			if (ck != null) {
				addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
				return ck.toString();
			} else {
				// 这里向平台发送请求，检查用户数据是否存在
				JSONObject userInfo = validUserWithPlatform(message);
				if (userInfo != null) {
					return failMsg(message.get("msgid"),StringValues.USER_REGISTRATION_IN_OHER_CENTER,ResponseType.USER_EXIST.toString());
				} else {
					Map<String, Object> params = new HashMap<>();
					params.put("status", Constant.SUCCESS);
					params.put("message_id", message.get("msgid"));
					if(messageDao.selectByParam(params)==0){
						// 如果平台没有，则本地持久化然后提交注册请求
						// 追加appid和secretkey
						addAppidAndSecretkey(message);
						// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
						producer.send(new KeyedMessage<String, String>(TOPIC,message.toString()));
						// 数据持久化
						addUser(message);
						log.info("###RegisterProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
						addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
						return successMsg(message.get("msgid"));
					}else{
						return failMsg(message.get("msgid"),StringValues.MESSAGE_SENDING_REPEAT);
					}
				}
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"),e.toString());
		}
	}
	
	/**
	 * 更新处理
	 * @param message
	 * @return
	 */
	private String updateProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = UserValid.validUpdate(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
				if (ResponseType.UUID_NOT_FOUND.toString().equals(ck.getString("code"))) {
					// 向平台发送请求并返回响应到客户端
					// 这里向平台发送请求，检查用户数据是否存在
					JSONObject userInfo = validUserWithPlatform(ceateValidJson(message));
					if (userInfo != null) {
						// 追加appid和secretkey
						addAppidAndSecretkey(message);
						// 当平台存在该用户的情况下执行平台的更新操作
						producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
						log.info("###UpdateProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
						addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
						return successMsg(message.get("msgid"));
					}else{
						// 如果平台也没有该用户返回错误信息
						return failMsg(message.get("msgid"),StringValues.USER_NOT_EXIST,ResponseType.USER_NOT_EXIST.toString());
					}
				} else {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					return ck.toString();
				}
			}
			Map<String, Object> params = new HashMap<>();
			params.put("status", Constant.SUCCESS);
			params.put("message_id", message.getString("msgid"));
			if(messageDao.selectByParam(params)==0){
				// 追加appid和secretkey
				addAppidAndSecretkey(message);
				// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
				producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
				// 如果校验通过，则用户在本中心存在且其他信息没有重复，执行更新用户
				updateUser(message);
				log.info("###UpdateProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
				addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
				return successMsg(message.get("msgid"));
			}else{
				return failMsg(message.get("msgid"),StringValues.MESSAGE_SENDING_REPEAT);
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}
	
	/**
	 * 黑名单处理
	 * @param message
	 * @return
	 */
	private String blackListProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = UserValid.validBlackList(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
				if (ResponseType.UUID_NOT_FOUND.toString().equals(
						ck.getString("code"))) {
					// 向平台发送请求并返回响应到客户端
					// 这里向平台发送请求，检查用户数据是否存在
					JSONObject userInfo = rebuildJSON(validUserWithPlatform(ceateValidJson(message)));
					if (userInfo != null) {
						if(userInfo.getInt("status")==Constant.USERCLOSE){
							return failMsg(message.get("msgid"),StringValues.USER_CAN_NOT_BLACKLIST);
						} else if(userInfo.getInt("status")==Constant.USERBLACKLIST){
							return failMsg(message.get("msgid"),StringValues.USER_HAVE_BEEN_BLACKLIST);
						}else{
							// 重构发送的json以符合平台数据格式
							message = rebuildSendJson(message,Constant.USERBLACKLIST);
							// 追加appid和secretkey
							addAppidAndSecretkey(message);
							// 当平台存在该用户的情况下执行平台的黑名单操作
							producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
							log.info("###lossUserProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
							addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
							return successMsg(message.get("msgid"));
						}
					}else{
						// 如果平台也没有该用户返回错误信息
						return failMsg(message.get("msgid"),StringValues.USER_NOT_EXIST,ResponseType.USER_NOT_EXIST.toString());
					}
				} else {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					return ck.toString();
				}
			}
			Map<String, Object> params = new HashMap<>();
			params.put("status", Constant.SUCCESS);
			params.put("message_id", message.get("msgid"));
			if(messageDao.selectByParam(params)==0){
				// 追加appid和secretkey
				addAppidAndSecretkey(message);
				// 如果校验通过，则用户在本中心存在,执行用户黑名单操作
				Map<String,String> query = new HashMap<>();
				query.put("uuid", message.getJSONObject("datas").getString("uuid"));
				User user = userDao.selectByCondition(query);
				if(user!=null && user.getEnabled()==Constant.USERCLOSE){
					return failMsg(message.get("msgid"),StringValues.USER_CAN_NOT_BLACKLIST);
				}else{
					message = rebuildSendJson(message,Constant.USERBLACKLIST);
					// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
					producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
					user.setEnabled(Constant.USERBLACKLIST);
					//更新用户状态为黑名单
					userDao.updateByUUID(user);
					//更新用户下绑定的所有卡的状态为注销
//					iccardDao.updateStatusByUUID(String.valueOf(ACCOUNTUNBIND), user.getUuid());
				}
				log.info("###lossUserProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
				addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
				return successMsg(message.get("msgid"));
			}else{
				return failMsg(message.get("msgid"),StringValues.MESSAGE_SENDING_REPEAT);
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}
	
	/**
	 * 激活处理
	 * @param message
	 * @return
	 */
	private String activateUserProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = UserValid.validActivate(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
				if (ResponseType.UUID_NOT_FOUND.toString().equals(
						ck.getString("code"))) {
					// 向平台发送请求并返回响应到客户端
					// 这里向平台发送请求，检查用户数据是否存在
					JSONObject userInfo = rebuildJSON(validUserWithPlatform(ceateValidJson(message)));
					if (userInfo != null) {
						if(userInfo.getInt("status")==Constant.USERCLOSE){
							return failMsg(message.get("msgid"),StringValues.USER_CAN_NOT_ACTIVATE);
						} else if(userInfo.getInt("status")==Constant.USERACTIVATE){
							return failMsg(message.get("msgid"),StringValues.USER_HAVE_BEEN_ACTIVATE);
						}else{
							// 重构发送的json以符合平台数据格式
							message = rebuildSendJson(message,Constant.USERACTIVATE);
							// 追加appid和secretkey
							addAppidAndSecretkey(message);
							// 当平台存在该用户的情况下执行平台的激活操作
							producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
							log.info("###activateUserProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
							addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
							return successMsg(message.get("msgid"));
						}
					}else{
						// 如果平台也没有该用户返回错误信息
						return failMsg(message.get("msgid"),StringValues.USER_NOT_EXIST,ResponseType.USER_NOT_EXIST.toString());
					}
				} else {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					return ck.toString();
				}
			}
			// 追加appid和secretkey
			addAppidAndSecretkey(message);
			Map<String, Object> params = new HashMap<>();
			params.put("status", Constant.SUCCESS);
			params.put("message_id", message.get("msgid"));
			if(messageDao.selectByParam(params)==0){
				// 如果校验通过，则用户在本中心存在,执行用户激活操作
				Map<String,String> query = new HashMap<>();
				query.put("uuid", message.getJSONObject("datas").getString("uuid"));
				User user = userDao.selectByCondition(query);
				if(user!=null && user.getEnabled()==Constant.USERCLOSE){
					return failMsg(message.get("msgid"),StringValues.USER_CAN_NOT_ACTIVATE);
				}else{
					message = rebuildSendJson(message,Constant.USERACTIVATE);
					// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
					producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
					user.setEnabled(Constant.USERACTIVATE);
					//更新用户状态为激活
					userDao.updateByUUID(user);
					//更新用户下绑定的所有卡的状态为可用
//					iccardDao.updateStatusByUUID(String.valueOf(ACCOUNTBIND), user.getUuid());
				}
				log.info("###activateUserProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
				addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
				return successMsg(message.get("msgid"));
			}else{
				return failMsg(message.get("msgid"),StringValues.MESSAGE_SENDING_REPEAT);
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}
	
	/**
	 * 用户注销处理
	 * @param message
	 * @return
	 */
	private String closeUserProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = UserValid.validClose(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
				if (ResponseType.UUID_NOT_FOUND.toString().equals(
						ck.getString("code"))) {
					// 向平台发送请求并返回响应到客户端
					// 这里向平台发送请求，检查用户数据是否存在
					JSONObject userInfo = rebuildJSON(validUserWithPlatform(ceateValidJson(message)));
					if (userInfo != null) {
						if(userInfo.getInt("status")==Constant.USERCLOSE){
							return failMsg(message.get("msgid"),StringValues.USER_HAVE_BEEN_CLOSE);
						}else{
							// 重构发送的json以符合平台数据格式
							message = rebuildSendJson(message,Constant.USERCLOSE);
							// 追加appid和secretkey
							addAppidAndSecretkey(message);
							// 当平台存在该用户的情况下执行平台的注销操作
							producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
							log.info("###closeUserProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
							addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
							return successMsg(message.get("msgid"));
						}
					}else{
						// 如果平台也没有该用户返回错误信息
						return failMsg(message.get("msgid"),StringValues.USER_NOT_EXIST,ResponseType.USER_NOT_EXIST.toString());
					}
				} else {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					return ck.toString();
				}
			}
			Map<String, Object> params = new HashMap<>();
			params.put("status", Constant.SUCCESS);
			params.put("message_id", message.get("msgid"));
			if(messageDao.selectByParam(params)==0){
				// 追加appid和secretkey
				addAppidAndSecretkey(message);
				// 如果校验通过，则用户在本中心存在,执行用户注销操作
				Map<String,String> query = new HashMap<>();
				query.put("uuid", message.getJSONObject("datas").getString("uuid"));
				User user = userDao.selectByCondition(query);
				if(user!=null && user.getEnabled()==Constant.USERCLOSE){
					return failMsg(message.get("msgid"),StringValues.USER_HAVE_BEEN_CLOSE);
				}else{
					message = rebuildSendJson(message,Constant.USERCLOSE);
					// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
					producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
					user.setEnabled(Constant.USERCLOSE);
					//更新用户状态为注销
					userDao.updateByUUID(user);
					//更新用户下绑定的所有卡的状态为不可用
					iccardDao.updateStatusByUUID(String.valueOf(Constant.ACCOUNTUNBIND), user.getUuid());
				}
				log.info("###closeUserProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
				addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
				return successMsg(message.get("msgid"));
			}else{
				return failMsg(message.get("msgid"),StringValues.MESSAGE_SENDING_REPEAT);
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}
	
	/**
	 * 构造用户查询参数
	 * @param message
	 * @return
	 */
	private JSONObject ceateValidJson(JSONObject message) {
		JSONObject validJson = new JSONObject();
		JSONObject datas = message.getJSONObject("datas");
		JSONObject result = new JSONObject();
		validJson.put("uuid", datas.getString("uuid"));
		validJson.put("loginname", "");
		validJson.put("agedness_card", "");
		validJson.put("mobile", "");
		validJson.put("email", "");
		validJson.put("sid", "");
		validJson.put("true_name", "");
		result.put("datas", validJson);
		return result;
	}

	/**
	 * register json数据处理方法
	 * 
	 * @param message
	 * @return
	 */
	private void addUser(JSONObject message) throws Exception {
		JSONObject userData = message.getJSONObject("datas");
		User user = createEntity(userData);
		userDao.insert(user);
		sqlSession.commit();
	}
	
	/**
	 * updatejson数据处理方法
	 * 
	 * @param message
	 * @return
	 */
	private void updateUser(JSONObject message) throws Exception {
		JSONObject userData = message.getJSONObject("datas");
		User user = createEntity(userData);
		userDao.updateByUUID(user);
		sqlSession.commit();
	}

	/**
	 * 创建实体类
	 * 
	 * @param userData
	 * @return
	 */
	private User createEntity(JSONObject userData) throws Exception{
		User user = new User();
		user.setUuid(userData.getString("uuid"));
		user.setLoginName(userData.getString("loginname"));
		user.setPhone(userData.getString("phone"));
		user.setHealthcareCard(userData.getString("healthcare_card"));
		user.setAvatar(userData.getString("avatar"));
		user.setWeixin(userData.getString("weixin"));
		user.setTrueName(userData.getString("true_name"));
		user.setPassword(userData.getString("passwd"));
		user.setBirthday(userData.getString("birthday"));
		user.setAgednessCard(userData.getString("agedness_card"));
		user.setQq(userData.getString("qq"));
		user.setAreaInfo(userData.getString("area_info"));
		user.setMobile(userData.getString("mobile"));
		user.setCommunity(userData.getString("community"));
		user.setContact(userData.getString("contact"));
		user.setSpouseName(userData.getString("spouse_name"));
		user.setEmail(userData.getString("email"));
		user.setSpouseStatus(userData.getString("spouse_status"));
		user.setAddress(userData.getString("address"));

		user.setConstellation(Integer.valueOf(userData.getString("constellation")));
		user.setUserClass(Integer.valueOf(userData.getString("user_class")));
		user.setPolitics(Integer.valueOf(userData.getString("politics")));
		user.setArea(Integer.valueOf(userData.getString("area")));
		user.setZodiac(Integer.valueOf(userData.getString("zodiac")));
		user.setNation(Integer.valueOf(userData.getString("nation")));
//		user.setIntegral(Integer.valueOf(userData.getString("integral")));
		user.setServiceClass(Integer.valueOf(userData.getString("service_class")));
		user.setMarriage(Integer.valueOf(userData.getString("marriage")));
		user.setProvince(Integer.valueOf(userData.getString("province")));
		user.setNationality(Integer.valueOf(userData.getString("nationality")));
		user.setCity(Integer.valueOf(userData.getString("city")));
		user.setSource(Integer.valueOf(userData.getString("source")));
		user.setGender(Integer.valueOf(userData.getString("gender")));
		user.setSid(userData.getString("sid"));
		user.setEnabled(Constant.USERACTIVATE);
		return user;
	}

	/**
	 * 重构发送数据
	 * @param message
	 * @param status
	 * @return
	 */
	private JSONObject rebuildSendJson(JSONObject message,Integer status){
		JSONObject datas = message.getJSONObject("datas");
		datas.put("status", status);
		message.put("datas", datas);
		return message;
	}

}